package cn.doitedu.demo.stream_table;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.Collector;

public class changelog表转changelog流 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射kafka中的ss-1
        tenv.executeSql("CREATE TABLE score_mysql (\n" +
                "     id INT,                     \n" +
                "     name string,                \n" +
                "     gender STRING,              \n" +
                "     score double,               \n" +
                "     PRIMARY KEY(id) NOT ENFORCED\n" +
                "     ) WITH (                   \n" +
                "     'connector' = 'mysql-cdc', \n" +
                "     'hostname' = 'doitedu',    \n" +
                "     'port' = '3306',           \n" +
                "     'username' = 'root',       \n" +
                "     'password' = 'root',       \n" +
                "     'database-name' = 'flinktest',\n" +
                "     'table-name' = 'flink_score')");

        Table table = tenv.from("score_mysql");
        //DataStream<Row> dataStream = tenv.toDataStream(table);  // score_mysql不是一个appendOnly流的表，所以，不能这样转
        DataStream<Row> changelogStream = tenv.toChangelogStream(table);
        SingleOutputStreamOperator<String> resultStream = changelogStream.process(new ProcessFunction<Row, String>() {
            @Override
            public void processElement(Row row, ProcessFunction<Row, String>.Context context, Collector<String> collector) throws Exception {

                Integer id = row.getFieldAs("id");
                String name = row.getFieldAs("name");
                String gender = row.getFieldAs("gender");

                // changelog流的row中，包含有 -U、+U、-D、+I 等操作类型
                RowKind kind = row.getKind();
                String s = kind.shortString();

                collector.collect(s + "," + id + "," + name + "," + gender);

            }
        });


        resultStream.print();

        env.execute();


    }

}
